当使用场景中不关心整个结果集的顺序时,可以使用并发导出数据功能以更快的速度将命中的数据全部返回。
前提条件
参数
参数 | 说明 | |
---|---|---|
tableName | 数据表名称。 | |
indexName | 多元索引名称。 | |
scanQuery | query | 多元索引的查询语句。支持精确查询、模糊查询、范围查询、地理位置查询、嵌套查询等,功能和Search接口一致。 |
limit | 扫描数据时一次能返回的数据行数。 | |
maxParallel | 最大并发数。请求支持的最大并发数由用户数据量决定。数据量越大,支持的并发数越多,每次任务前可以通过ComputeSplits API进行获取。 | |
currentParallelId | 当前并发ID。取值范围为[0, maxParallel)。 | |
token | 用于翻页功能。ParallelScan请求结果中有下一次进行翻页的token,使用该token可以接着上一次的结果继续读取数据。 | |
aliveTime | ParallelScan的当前任务有效时间,也是token的有效时间。默认值为60,建议使用默认值,单位为秒。如果在有效时间内没有发起下一次请求,则不能继续读取数据。持续发起请求会刷新token有效时间。
说明 由于服务端采用异步方式清理过期任务,因此当前任务只保证在设置的有效时间内不会过期,但不能保证有效时间之后一定过期。
|
|
columnsToGet | ParallelScan目前仅可以扫描多元索引中的数据,需要在创建多元索引时设置附加存储(即store=true)。 | |
sessionId | 本次并发扫描数据任务的sessionId。创建Session可以通过ComputeSplits API来创建,同时获得本次任务支持的最大并发数。 |
示例
//1.获取sessionId。
let computeSplits = await new Promise((resolve, reject) => {
client.computeSplits({
tableName: tableName,
searchIndexSplitsOptions: {
indexName: indexName,
}
}, function (err, data) {
if (err) {
console.log('computeSplits error:', err.toString());
reject(err);
} else {
console.log('computeSplits success:', data);
resolve(data)
}
})
})
//2.构造query。
const scanQuery = {
query: {
queryType: TableStore.QueryType.MATCH_ALL_QUERY,
},
limit: 1000,
aliveTime: 30,
token: undefined,
currentParallelId: 0,
maxParallel: 1,
}
//3.构造ParallelScan请求(该示例为了方便介绍使用同步请求进行展示,实际业务中可修改为异步)。
const parallelScanPromise = function () {
return new Promise(function (resolve, reject) {
client.parallelScan({
tableName: tableName,
indexName: indexName,
columnToGet: {
returnType: TableStore.ColumnReturnType.RETURN_ALL_FROM_INDEX,
},
sessionId: computeSplits.sessionId,
scanQuery: scanQuery,
}, function (err, data) {
if (err) {
console.log('parallelScan error:', err.toString());
reject(err);
} else {
console.log("parallelScan, rows:", data.rows.length)
resolve(data)
}
});
})
}
let totalCount = 0 //示例代码记录总行数。
let parallelScanResponse = await parallelScanPromise()
totalCount = totalCount + parallelScanResponse.rows.length
//4.迭代拉取数据,直到拉取所有数据结束。
while (parallelScanResponse.nextToken !== null && parallelScanResponse.nextToken.length > 0) {
scanQuery.token = parallelScanResponse.nextToken
parallelScanResponse = await parallelScanPromise()
totalCount += parallelScanResponse.rows.length
}
console.log("total rows:", totalCount)